package com.yswy.common.utils.bloom;

import com.google.common.base.Preconditions;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author Rocky
 * @version 1.0
 * @description
 * @email inaho00@foxmail.com
 * @createDate 2021/6/10 下午3:11
 * @log
 */
@Component
public class BloomRedisService {

	private RedisTemplate<Object, Object> redisTemplate;

	private BloomFilterHelper bloomFilterHelper;

	public void setBloomFilterHelper(BloomFilterHelper bloomFilterHelper) {
		this.bloomFilterHelper = bloomFilterHelper;
	}

	public void setRedisTemplate(RedisTemplate<Object, Object> redisTemplate) {
		this.redisTemplate = redisTemplate;
	}

	/**
	 * 根据给定的布隆过滤器添加值
	 */
	public <T> void addByBloomFilter(String key, T value) {
		Preconditions.checkArgument(bloomFilterHelper != null, "bloomFilterHelper不能为空");
		//数据写回redis
		SessionCallback<Object> callback = new SessionCallback<Object>() {
			@Override
			public Object execute(RedisOperations operations) throws DataAccessException {
				operations.multi();
				int[] offset = bloomFilterHelper.murmurHashOffset(value);
				for (int i : offset) {
					operations.opsForValue().setBit(key, i, true);
				}
				return operations.exec();
			}
		};
		redisTemplate.execute(callback);
	}

	public <T> void multiAddByBloomFilter(String key,List<T> value) {
		Preconditions.checkArgument(bloomFilterHelper != null, "bloomFilterHelper不能为空");
		//数据写回redis
		SessionCallback<Object> callback = new SessionCallback<Object>() {
			@Override
			public Object execute(RedisOperations operations) throws DataAccessException {
				operations.multi();
				value.forEach(e->{
					int[] offset = bloomFilterHelper.murmurHashOffset(e);
					for (int i : offset) {
						operations.opsForValue().setBit(key, i, true);
					}
				});
				return operations.exec();
			}
		};
		redisTemplate.execute(callback);
	}

	/**
	 * 根据给定的布隆过滤器添加值
	 */
	public <T> void deleteByBloomFilter(String key, T value) {
		Preconditions.checkArgument(bloomFilterHelper != null, "bloomFilterHelper不能为空");
		int[] offset = bloomFilterHelper.murmurHashOffset(value);
		for (int i : offset) {
			redisTemplate.opsForValue().setBit(key, i, false);
		}
	}

	/**
	 * 根据给定的布隆过滤器判断值是否存在
	 */
	public <T> boolean includeByBloomFilter(String key, T value) {
		Preconditions.checkArgument(bloomFilterHelper != null, "bloomFilterHelper不能为空");
		int[] offset = bloomFilterHelper.murmurHashOffset(value);
		for (int i : offset) {
			if (!redisTemplate.opsForValue().getBit(key, i)) {
				return false;
			}
		}
		return true;
	}
}
